#ifndef __M_CHANNEL_H__
#define __M_CHANNEL_H__

#include "../mqcommon/mq_helper.hpp"
#include "../mqcommon/mq_logger.hpp"
#include "../mqcommon/mq_msg.pb.h" 
#include "../mqcommon/mq_proto.pb.h"
#include "../mqcommon/mq_threadPool.hpp"
#include "muduo/net/TcpServer.h"
#include "muduo/proto/codec.h"
#include "mq_consumer.hpp"
#include "mq_vhost.hpp"
#include "mq_route.hpp"
#include <iostream>
#include <unordered_map>
#include <memory>
#include <mutex>
#include <string>

namespace wsmq
{
    using openChannelRequestPtr = std::shared_ptr<openChannelRequest>;
    using closeChannelRequestPtr = std::shared_ptr<closeChannelRequest>;
    using declareExchangeRequestPtr = std::shared_ptr<declareExchangeRequest>;
    using removeExchangeRequestPtr = std::shared_ptr<removeExchangeRequest>;
    using declareQueueRequestPtr = std::shared_ptr<declareQueueRequest>;
    using removeQueueRequestPtr = std::shared_ptr<removeQueueRequest>;
    using queueBindRequestPtr = std::shared_ptr<queueBindRequest>;
    using QueueUnBindRequestPtr = std::shared_ptr<QueueUnBindRequest>;
    using basicPublishRequestPtr = std::shared_ptr<basicPublishRequest>;
    using basicConsumeRequestPtr = std::shared_ptr<basicConsumeRequest>;
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    using basicCancelRequestPtr = std::shared_ptr<basicCancelRequest>;
    using basicAckRequestPtr = std::shared_ptr<basicAckRequest>;
    class Channel
    {
    public:
        using ptr = std::shared_ptr<Channel>;
        Channel(const std::string& cid, 
                const muduo::net::TcpConnectionPtr conn, 
                const VirtualHost::ptr host, 
                const ConsumerManager::ptr cmp, 
                const ProtobufCodecPtr& codec,
                const ThreadPool::ptr pool):
                _cid(cid), 
                _conn(conn),
                _host(host),
                _cmp(cmp),
                _codec(codec),
                _pool(pool)
        {
            DLOG("new a channel: %p", this);
        }
        ~Channel()
        {
            DLOG("delete a channel: %p", this);
            if(_consumer.get() != nullptr)
            {
                _cmp->remove(_consumer->_tag, _consumer->_qname);
            }
        }
        void declareExchange(const declareExchangeRequestPtr& req)
        {
            bool ret = _host->declareExchange(req->exchange_name(), req->type(),
                req->durable(), req->auto_delete(), req->args());

            basicResponse(req->cid(), req->rid() , ret);
        }
        void removeExchange(const removeExchangeRequestPtr& req)
        {
            _host->deleteExchange(req->exchange_name());
            basicResponse(req->cid(), req->rid() , true);
        }

        void declareQueue(const declareQueueRequestPtr& req)
        {
            bool ret = _host->declareQueue(req->queue_name(), req->durable(), req->exclusive(),
                req->auto_delete(), req->args());

            _cmp->initQueueConsumer(req->queue_name());
            basicResponse(req->cid(), req->rid() , ret);
        }
        void removeQueue(const removeQueueRequestPtr& req)
        {
            _host->deleteQueue(req->queue_name());
            _cmp->destroyQueueConsumer(req->queue_name());
            basicResponse(req->cid(), req->rid(), true);
        }

        void queueBind(const queueBindRequestPtr& req)
        {
            bool ret = _host->bind(req->exchange_name(), req->queue_name(), req->binding_key());
            basicResponse(req->cid(), req->rid(), ret);
        }
        void QueueUnBind(const QueueUnBindRequestPtr& req)
        {
            _host->unBind(req->exchange_name(), req->queue_name());
            basicResponse(req->cid(), req->rid(), true);
        }

        // 进行消息发布
        void basicPublish(const basicPublishRequestPtr& req)
        {
            // 判断交换机是否存在
            auto ep = _host->selectExchange(req->exchange_name());
            if(ep.get() == nullptr)
            {
                basicResponse(req->cid(), req->rid(), false);
            }
            // 进行交换路由，判断消息可以发布到哪些队列中
            MsgQueueBindingMap mqbm = _host->exchangeBindings(req->exchange_name());
            BasicProperties* bp = nullptr;
            std::string routing_key;
            if(req->has_properties())
            {
                bp = req->mutable_properties();
                routing_key = bp->routing_key();
            }
            for(auto& binding : mqbm)
            {
                if(Router::route(ep->type, routing_key, binding.second->_binding_key))
                {
                    // 将消息添加到队列中
                    _host->basicPublish(binding.first, bp, req->body());
                    // 向线程池添加消息消费任务（向指定队列的订阅者发布消息）
                    _pool->Push(std::bind(&Channel::consume, this, binding.first));
                }
            }
            basicResponse(req->cid(), req->rid(), true);
        }

        // 进行消息确认
        void basicAck(const basicAckRequestPtr& req)
        {
            _host->basicAck(req->queue_name(), req->msg_id());
            basicResponse(req->cid(), req->rid(), true);
        }

        // 进行队列订阅
        void basicConsume(const basicConsumeRequestPtr& req)
        {
            // 判断队列存在
            bool ret = _host->existQueue(req->queue_name());
            if(ret == false)
            {
                basicResponse(req->cid(), req->rid(), false);
            }
            // 创建队列消费者，此时该channel定位就是消费者
            auto cb = std::bind(&Channel::callback, this, std::placeholders::_1, 
                std::placeholders::_2, std::placeholders::_3);
            _consumer = _cmp->insert(req->consumer_tag(), req->queue_name(), 
                req->auto_ack(), cb);
            basicResponse(req->cid(), req->rid(), true);
        }
        void basicCancel(const basicCancelRequestPtr& req)
        {
            _cmp->remove(req->consumer_tag(), req->queue_name());
            basicResponse(req->cid(), req->rid(), true);
        }
    private:
        void callback(const std::string& tag, const BasicProperties* bp, const std::string& body)
        {
            // 将传入的参数构建为响应，推送到客户端
            basicConsumeResponse resp;
            resp.set_cid(_cid);
            resp.set_consumer_tag(tag);
            resp.set_body(body);
            if(bp)
            {
                resp.mutable_bp()->set_id(bp->id());
                resp.mutable_bp()->set_delivery_mode(bp->delivery_mode());
                resp.mutable_bp()->set_routing_key(bp->routing_key());
            }
            _codec->send(_conn, resp);
        }
        void consume(const std::string& qname)
        {
            // 从队列选出一条消息
            MessagePtr mp = _host->basicConsume(qname);
            if(mp.get() == nullptr)
            {
                DLOG("线程池执行消费任务失败，队列 %s 没有消息", qname.c_str());
                return ;
            }
            // 从订阅队列的订阅者中选出一个
            Consumer::ptr cp = _cmp->choose(qname);
            if(cp.get() == nullptr)
            {
                DLOG("线程池执行消费任务失败，队列 %s 没有订阅者", qname.c_str());
                return ;
            }
            // 调用订阅者的对应的消息处理函数，实现消息的推送
            cp->_callback(cp->_tag, mp->mutable_payload()->mutable_properties(), mp->payload().body());
            // 如果订阅者是自动确认，则直接确认
            if(cp->_auto_ack) _host->basicAck(qname, mp->payload().properties().id());
        }
        void basicResponse(const std::string& cid, const std::string& rid, bool ret)
        {
            basicCommonResponse resp;
            resp.set_cid(cid);
            resp.set_rid(rid);
            resp.set_ok(ret);
            _codec->send(_conn, resp);
        }
    private:
        std::string _cid;
        Consumer::ptr _consumer;
        muduo::net::TcpConnectionPtr _conn;
        VirtualHost::ptr _host;
        ConsumerManager::ptr _cmp;
        ProtobufCodecPtr _codec;
        ThreadPool::ptr _pool;
    };
    class ChannelManager
    {
    public:
        using ptr = std::shared_ptr<ChannelManager>;
        ChannelManager(){}
        bool openChannel(const std::string cid, 
            const muduo::net::TcpConnectionPtr conn, 
            const VirtualHost::ptr host, 
            const ConsumerManager::ptr cmp, 
            const ProtobufCodecPtr& codec,
            const ThreadPool::ptr pool)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channels.find(cid);
            if(it != _channels.end())
            {
                DLOG("信道 %s 已存在", cid.c_str());
                return false;
            }
            Channel::ptr channel = std::make_shared<Channel>(cid, 
                conn, host, cmp, codec, pool);
            _channels.insert(std::make_pair(cid, channel));
            return true;
        }
        void closeChannel(const std::string cid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _channels.erase(cid);
        }
        Channel::ptr getChannel(const std::string& cid)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channels.find(cid);
            if(it == _channels.end())
            {
                return Channel::ptr();
            }
            return it->second;
        }
    private:
        std::mutex _mutex;
        std::unordered_map<std::string, Channel::ptr> _channels;
    };
}

#endif